from pyspark.sql import SparkSession
import os
import pyspark.sql.functions as F

from cn.itcast.tag.base.BaseModel import BaseModel
from cn.itcast.tag.bean.ESMeta import ruleToESMeta

"""
-------------------------------------------------
   Description :	TODO：
   SourceFile  :	ConsumerCycleModel
   Author      :	itcast team
-------------------------------------------------
"""

# 0.设置系统环境变量
os.environ['JAVA_HOME'] = '/export/server/jdk1.8.0_241/'
os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/envs/pyspark_env/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/envs/pyspark_env/bin/python3'

# todo 消费周期
class ConsumerCycleModel(BaseModel):
    def compute(self, es_df, five_df):
        """
        标签计算主逻辑：基于用户的最近一次下单时间，计算其处于何种消费周期
        
        Args:
            es_df: 从ES读取的业务数据（订单数据）
            five_df: 五级标签规则数据
            
        Returns:
            DataFrame: 包含用户ID和对应标签ID的结果数据框
        """
        # 按照memberid分组，获取每个用户的最近一次下单时间
        es_df = es_df.groupBy("memberid").agg(F.max("finishtime").alias("finishtime"))
        
        # 计算用户最近一次下单距今的天数
        es_df = es_df.select(
            "memberid",
            F.current_date().alias("current_date"),  # 获取当前日期
            F.from_unixtime("finishtime", format='yyyy-MM-dd').alias("finishtime"),  # 将时间戳转换为标准日期格式
            F.datediff(F.current_date(), F.from_unixtime("finishtime", format='yyyy-MM-dd')).alias("days")  # 计算时间差
        )
        
        # 只保留需要的字段：用户ID和天数
        es_df = es_df.select("memberid", "days")
        
        # 展示中间结果（调试用）
        es_df.show()

        #split(str,pattern)：切割函数
        five_df = five_df.select("id",
                       F.split("rule","-")[0].alias("start"),
                       F.split("rule","-")[1].alias("end"))
        five_df.show()
        #join中带有between的写法(start,end)
        new_df = es_df.join(five_df,es_df['days'].between(five_df['start'],five_df['end']),'left')\
            .select(es_df['memberid'].alias("userId"),five_df['id'].alias("tagsId"))
        new_df.show()
        return new_df

if __name__ == '__main__':
    model = ConsumerCycleModel(23)
    model.execute()
